package org.kafka.connector;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.source.SourceConnector;

public class MyConnector extends SourceConnector {


    public static String FILE_CONFIG = "filename";
    public static String TOPIC = "topic";
    private String filename;
    private String topic;
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
        .define(FILE_CONFIG, ConfigDef.Type.STRING, Importance.HIGH, "Source filename.")
        .define(TOPIC, ConfigDef.Type.STRING, Importance.HIGH, "The topic to publish data to");

    public String getFilename() {
        return filename;
    }

    public void setFilename(String filename) {
        this.filename = filename;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String version() {
        return null;
    }

    public void start(Map<String, String> map) {
        filename = map.get(FILE_CONFIG);
        topic = map.get(TOPIC);
    }

    public Class<? extends Task> taskClass() {
        return MyTask.class;
    }

    public List<Map<String, String>> taskConfigs(int maxTasks) {
        //高级用法 这里可以动态扩容
        ArrayList<Map<String, String>> configs = new ArrayList<>();
        String[] file = filename.split(",");
        // Only one input partition makes sense.
        Arrays.stream(file).forEach(
            f -> {
                Map<String, String> config = new HashMap<>();
                if (f != null) {
                    config.put(FILE_CONFIG, f);
                }
                config.put(TOPIC, topic);
                configs.add(config);
            }
        );

        return configs;
    }

    public void stop() {

    }

    @Override
    public ConfigDef config() {
        return CONFIG_DEF;
    }
}
